package hydrap2p.protocol.ftp;

import hydrap2p.ServiceLocation;
import hydrap2p.download.DownloadProtocol;
import hydrap2p.download.Downloader;
import hydrap2p.download.Source;
import hydrap2p.download.DownloadListener.newDownload_args;
import hydrap2p.helper.ServiceConnector;
import hydrap2p.helper.ThriftServerThread;
import hydrap2p.helper.ThriftUtils;
import hydrap2p.library.FileNotFound;
import hydrap2p.library.IOProblem;
import hydrap2p.library.InvalidUID;
import hydrap2p.library.Library;
import hydrap2p.library.Range;
import hydrap2p.logging.Level;
import hydrap2p.logging.Logging;
import hydrap2p.modulemanager.ModuleControl;
import hydrap2p.servicemanager.NameTaken;
import hydrap2p.speedlimiter.SpeedLimiter;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.MalformedURLException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.log4j.PropertyConfigurator;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;

/**
 * The Class FtpProtocol.
 */
public class FtpProtocol implements DownloadProtocol.Iface, ModuleControl.Iface {
	//Init logger
	static {
		//Configure log4j
		Properties log4jProps = new Properties();
		log4jProps.setProperty("log4j.rootLogger", "TRACE, console");
		log4jProps.setProperty("log4j.appender.console", "org.apache.log4j.ConsoleAppender");
		log4jProps.setProperty("log4j.appender.console.layout", "org.apache.log4j.PatternLayout");
		log4jProps.setProperty("log4j.appender.console.layout.ConversionPattern", "%-4r [%t] %-5p %c %x - %m%n");

		PropertyConfigurator.configure(log4jProps);
	}
	
	// Class string name, doubling as logging marker
	private static final Marker TAG = MarkerFactory.getDetachedMarker("FtpProtocol");

	// Console logger - used before connections established
	private static final Logger consoleLogger = LoggerFactory.getLogger(TAG.getName());

	private static ThriftServerThread t;
	private static ThriftServerThread controlthread;

	private int port;
	
	private Map<String, FTPClient> connections;

	/**
	 * Instantiates a new ftp protocol.
	 * 
	 * @param port the port
	 */
	public FtpProtocol(int port) {
		this.port = port;
		connections=new HashMap<String, FTPClient>();
	}

	/**
	 * The main method.
	 * 
	 * @param args the arguments
	 */
	public static void main(String[] args) {
		int port = 34656;
		FtpProtocol ftpProtocol = null;

		try {
			ftpProtocol = new FtpProtocol(port);
		} catch(Exception e) {
			System.out.println("Failed to start FtpProtocol, " + e.getMessage());
			System.exit(1);
		}

		// Start server
		consoleLogger.info(TAG, "Starting logging server");
		
		try {
			t = ThriftUtils.initThriftServer(new DownloadProtocol.Processor(ftpProtocol), "FtpProtocol", port);
		} catch (BindException e) {
			consoleLogger.error(TAG, "Failed to start FtpProtocol!  Could not bind to port.");
			e.printStackTrace();
			return;
		} catch (IOException e) {
			consoleLogger.error(TAG, "Failed to start FtpProtocol!");
			e.printStackTrace();
			return;
		} catch (NameTaken e) {
			consoleLogger.error(TAG, "Failed to start FtpProtocol!  Duplicate module exists.");
			e.printStackTrace();
			return;
		}
		try {
			controlthread = ThriftUtils.initThriftServer(new ModuleControl.Processor(ftpProtocol), "FtpProtocol.Control", port);
		} catch (BindException e) {
			consoleLogger.error(TAG, "Failed to start FtpProtocol!  Could not bind to port.");
			e.printStackTrace();
			return;
		} catch (IOException e) {
			consoleLogger.error(TAG, "Failed to start FtpProtocol!");
			e.printStackTrace();
			return;
		} catch (NameTaken e) {
			consoleLogger.error(TAG, "Failed to start FtpProtocol!  Duplicate module exists.");
			e.printStackTrace();
			return;
		}
		
		//Ensure module connections
		boolean manager = ThriftUtils.waitForManager();
		boolean library = ThriftUtils.waitForLibrary();
		boolean downloader = ThriftUtils.waitForDownloader();
		boolean logging = ThriftUtils.waitForLogging();
		
		if(manager && library && downloader) {
			try {
				Downloader.Client downloaderModule = ServiceConnector.getDownloader();
				downloaderModule.addDownloadProtocol(new ServiceLocation("localhost", port), "ftp");
			} catch (TException e) {
				consoleLogger.error(TAG, "Failed to register with downloader!");
				return;
			}
			
			t.start();
			controlthread.start();
		}
		else
			consoleLogger.error(TAG, "Failed to connect to essential modules!  Exiting.");
	}

	/* (non-Javadoc)
	 * @see hydrap2p.download.DownloadProtocol.Iface#cancelDownload(long, hydrap2p.download.Source)
	 */
	@Override
	public void cancelDownload(long uid, Source source) throws TException {
		// TODO Auto-generated method stub

	}
	
	public void download(long uid, Source source, Range range, Downloader.Iface downloader, Library.Iface library)
	{

		log(Level.DEBUG, "Incoming download request from "+source.ip);
		SpeedLimiter.Client speedlimiter = ServiceConnector.getSpeedLimiter();
		
		//Check source
		try {
			if(source == null || source.getProtocol() == null) {
				downloader.erroredDownloading(uid, source, range, true);
				log(Level.ERROR, "Invalid source passed to Ftp!");
				return;
			} else if(!source.getProtocol().equalsIgnoreCase("ftp")) {
				downloader.erroredDownloading(uid, source, range, true);
				log(Level.ERROR, "Source with incorrect protocol passed to Ftp!");
				return;
			}
		}
		catch (TException e)
		{
			log(Level.ERROR, "Downloader is not available, even after telling me what to do. Very strange.");
			return;
		}

		FTPClient conn = null;
		BufferedInputStream dlStream = null;
		
		int reply;

		try {
			synchronized(connections)
			{
				conn=connections.get(source.ip);
				if (conn==null)
				{
					log(Level.DEBUG, "Saving connection to "+source.ip+" in connection pool");
					conn = new FTPClient();
					conn.setDefaultTimeout(10000);
					connections.put(source.ip, conn);
				}
			}
			
			// connect to the server
			synchronized(conn)
			{
				if (!conn.isConnected())
				{
					try {
						try {
							conn.connect(source.ip, source.port);
						} catch(IOException e) {
							e.printStackTrace();
							log(Level.ERROR, "Could not connect to server at " + source.ip + ":" + source.port+" : "+e.getLocalizedMessage());
							downloader.erroredDownloading(uid, source, range, true);
							return;
						}
		
						reply = conn.getReplyCode();
		
						if(!FTPReply.isPositiveCompletion(reply)) {
							conn.disconnect();
							downloader.erroredDownloading(uid, source, range, true);
							System.err.println("FTP server refused connection.");
							return;
						}
		
						//Try to authenticate from Source, otherwise try anonymously
						String username = source.getExtra().get("username");
						String password = source.getExtra().get("password");
						
						//Login if set
						if(username != null && password != null) {
							boolean result = conn.login(username, password);
							//If failed, also try anonymous
							if(!result)
								conn.login("anonymous", "lolcats@hydrap2p.com");
						}
						else
							conn.login("anonymous", "lolcats@hydrap2p.com");
						
						// set other basic parameters
						conn.setFileType(FTP.BINARY_FILE_TYPE);
						
						String path = source.getExtra().get("path");
						if(path == null) {
							log(Level.ERROR, "Path extra does not exist in source!");
							downloader.erroredDownloading(uid, source, range, true);
							return;
						}
						
						//Set desired piece
						conn.setRestartOffset(range.start);
						
						//Check offset
						//TODO: Improve validation via more reliable means
						if(conn.getRestartOffset() != range.start) {
							downloader.erroredDownloading(uid, source, range, true);
							return;
						}
						
						conn.enterLocalPassiveMode();
					} catch(MalformedURLException e) {
						log(Level.ERROR, "Malformed url in source");
						e.printStackTrace();
						downloader.erroredDownloading(uid, source, range, true);
						return;
					}

				}

				//Get data
				int bytesRead = 0; 
				int offset = 0;
				String path = source.getExtra().get("path");
				dlStream = new BufferedInputStream(conn.retrieveFileStream(path));
				int loopCounter = 0;  //Used to throttle log output

				byte[] data = new byte[(int)range.len];
				ByteBuffer highdata = ByteBuffer.wrap(data);
				
				//Loop until we've gotten range.len bytes
				while (bytesRead < range.len && (offset = dlStream.read(data, bytesRead, data.length - bytesRead)) >= 0) { 
					bytesRead += offset;
					loopCounter++;
	
					if (speedlimiter!=null)
					{
						try {
							speedlimiter.attempt(source, true, offset);
						}
						catch (Exception e)
						{
							speedlimiter=null;
						}
					}
					
					try {
						
						library.saveData(uid, range.start+bytesRead-offset, ByteBuffer.wrap(data, bytesRead-offset, offset));
					} catch(FileNotFound e) {
						//Problem saving data
						e.printStackTrace();
					} catch(IOProblem e) {
						//Problem saving data
						e.printStackTrace();
					} catch (InvalidUID e) {
						//Problem saving data
						e.printStackTrace();
					}
					
					if(loopCounter % 200 == 0)
						log(Level.VERBOSE, "Download " + (float)bytesRead/(float)data.length * 100.0f + "% complete");
				}

				if(bytesRead != range.len) {
					log(Level.MISHAP, "Mismatch between bytes requested and bytes downloaded!");
				
					downloader.erroredDownloading(uid, source, range, false);
				}
				downloader.finishedDownloading(uid, source, range);
				//TODO: Verify data
				
				log(Level.INFO, "Successfully downloaded " + bytesRead + " bytes of data");
			}
			
//			conn.logout();
		} catch(IOException e) {
			//Something horrible happened during the download
			log(Level.ERROR, "IOException downloading data!");
			e.printStackTrace();
		} catch (TException e) {
			e.printStackTrace();
			log(Level.ERROR, "A communication error happened with a required module!");
			return;
		} finally {
			//Close stream
			if(dlStream != null) {
				try {
					dlStream.close();
				} catch(IOException e) {
					e.printStackTrace();
				}
			}

			//Close connection
			if(conn != null) {
				if(conn.isConnected()) {
					try {
						conn.disconnect();
					} catch(IOException e) {
						e.printStackTrace();
					}
				}
			}
		}
		
		return;
	}
	/* (non-Javadoc)
	 * @see hydrap2p.download.DownloadProtocol.Iface#download(long, hydrap2p.download.Source, hydrap2p.library.Range)
	 */
	@Override
	public void download(long uid, Source source, Range range) {
		//TODO Wait for each and check for validity
		Downloader.Client downloader = ServiceConnector.getDownloader();
		Library.Client library = ServiceConnector.getLibrary();
		download(uid, source, range, downloader, library);
	}

	/* (non-Javadoc)
	 * @see hydrap2p.Service.Iface#ping()
	 */
	@Override
	public void ping() throws TException {

	}

	/* (non-Javadoc)
	 * @see hydrap2p.Service.Iface#supportedInterfaces()
	 */
	@Override
	public List<String> supportedInterfaces() throws TException {
		return Arrays.asList("DownloadProtocol", "ModuleControl");
	}

	/**
	 * Log.
	 * 
	 * @param severity the severity
	 * @param logMessage the log message
	 */
	protected void log(Level severity, String logMessage) {
		Logging.Client logging = ServiceConnector.getLogger();
		
		//Log to Logging module if possible, and fall back to console.
		try {
			logging.log(severity, TAG.toString(), logMessage);
		} catch(TException e) {
			logToSystemError(severity, logMessage);
		} catch(NullPointerException e) {
		 	logToSystemError(severity, logMessage);
		}
	}

	/**
	 * Log to system error.
	 * 
	 * @param severity the severity
	 * @param logMessage the log message
	 */
	private void logToSystemError(Level severity, String logMessage) {
		//Log to system error.  If they can't be arsed to use a Logging constant as severity, the message gets logged at trace level
		switch(severity) {
		case VERBOSE:
			consoleLogger.trace(TAG, logMessage);
			break;
		case DEBUG:
			consoleLogger.debug(TAG, logMessage);
			break;
		case INFO:
			consoleLogger.info(TAG, logMessage);
			break;
		case MISHAP:
			consoleLogger.warn(TAG, logMessage);
			break;
		case ERROR:
			consoleLogger.error(TAG, logMessage);
			break;
		default:
			consoleLogger.trace(TAG, logMessage);
		}
	}

	/* (non-Javadoc)
	 * @see hydrap2p.manager.ModuleControl.Iface#stop()
	 */
	@Override
	public void stop() throws TException {
		log(Level.INFO, "Stopping FTP server");
		
		t.shutdown();
		controlthread.shutdown();
	}
}
